package zy.learn.demo.structuredstreaming.basic

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.Trigger

/**
 * 先启动应用，然后马上在cmd中运行 nc64 -l -p 9999
 * 待应用启动好后，可以在cmd中输入字符验证应用了
 * 或
 * 在Centos7中运行 ncat -v -lp 9999
 */
object WordCount3 {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")

    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("WordCount3").getOrCreate()

    // 1.从数据源加载数据
    val lines = spark.readStream
      .format("socket")
      .option("host", "192.168.122.200")
      .option("port", 9999)
      .load()

    import spark.implicits._

    // 2. 聚合计算
    lines.as[String].flatMap(_.split(" ")).createOrReplaceTempView("testTable")

    //    val wordCount = spark.sql("select value, count(*) as count from testTable group by value")
    val wordCount = spark.sql(
      """
        |select value, count(*) as count
        |  from testTable
        | group by value
        | """.stripMargin)

    // 3. 输出
    /**
     * Complete Mode: 整个更新的结果表会被写入到外部存储，必须有聚合才可以用此模式输出
     * Append Mode: 从上次触发结束开始算起，仅把那些追加到结果表中的行（不会改变的行）输出。
     * 如有聚合操作，必须添加watermark，没过期的聚合值不输出，只输出已过期的聚合值（这类聚合值不会变化）
     * Update Mode：从上次触发结束算起，仅输出结果表中更新的行（已改变的行）。从2.1.1可用。
     * 如果查询不包括聚合操作，则Update Mode 和 Append Mode相同。
     * Default Trigger：尽快运行微批处理
     * Trigger.Once: 只执行一次
     * Trigger.ProcessingTime：间隔达到100ms级，间隔时间段进行微批处理
     * Trigger.Continuous：连续执行，最小延迟为1ms级别，参数指定checkpoint的时间间隔
     */
    val result = wordCount.writeStream
      .format("console")
      .outputMode("update")
      //      .outputMode("complete")
      //      .outputMode("append")
      .trigger(Trigger.ProcessingTime("2 seconds"))
      //      .trigger(Trigger.Once())
      .start()

    result.awaitTermination()
  }
}
